package com.dtruth.dataflow;

import org.slf4j.Logger;

import com.dtruth.dataflow.bolt.AutoChannelBolt;
import com.dtruth.dataflow.bolt.BodyReformatBolt;
import com.dtruth.dataflow.bolt.DataNormalBolt;
import com.dtruth.dataflow.bolt.DataPushOutBolt;
import com.dtruth.dataflow.bolt.DataRefererBolt;
import com.dtruth.dataflow.bolt.DataStatisticBolt;
import com.dtruth.dataflow.bolt.DataValidateBolt;
import com.dtruth.dataflow.bolt.PicsProcessBolt;
import com.dtruth.dataflow.bolt.RmDuplctBolt;
import com.dtruth.dataflow.spout.RedisDataFlowSpout;
import com.dtruth.dataflow.util.MyLoggerFactory;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.topology.TopologyBuilder;

public class TopologyLocalCluster {
	// static {System.setProperty("logback.configurationFile", "logback.xml");}

	private static Logger log = MyLoggerFactory
			.getLogger(TopologyLocalCluster.class);

	// private static Logger log =
	// LoggerFactory.getLogger(DataFlowTopology.class);

	public static void main(String[] args) throws InterruptedException,
			AlreadyAliveException {
		// Topology definition

		TopologyBuilder builder = new TopologyBuilder();

		// 从RabbitMQ中读取数据,并判断是否满足处理条件
		builder.setSpout("data-reader", new RedisDataFlowSpout());

		// 格式标准化：发布日期、域提取、分类重构
		builder.setBolt("data-normal", new DataNormalBolt()).shuffleGrouping(
				"data-reader");

		// 通过referer从redis中读取信息：频道映射及新闻质量分
		builder.setBolt("data-referer", new DataRefererBolt()).shuffleGrouping(
				"data-normal");

		// 自动频道划分：划分到频道
		builder.setBolt("auto-channel", new AutoChannelBolt()).shuffleGrouping(
				"data-referer");

		// Body标签替换
		builder.setBolt("body-reformat", new BodyReformatBolt())
				.shuffleGrouping("auto-channel");

		// Body标签替换
		builder.setBolt("data-validate", new DataValidateBolt())
				.shuffleGrouping("body-reformat");

		// 图片统计
		builder.setBolt("data-statistic", new DataStatisticBolt())
				.shuffleGrouping("data-validate");

		// 去重处理
		builder.setBolt("rm-duplicate", new RmDuplctBolt()).shuffleGrouping(
				"data-validate");

		// 缩略图过滤
		builder.setBolt("pics-process", new PicsProcessBolt()).shuffleGrouping(
				"rm-duplicate");

		// 把结果推送到cms中
		builder.setBolt("data-pushout", new DataPushOutBolt()).shuffleGrouping(
				"pics-process");

		// Configuration
		Config conf = new Config();

		conf.setDebug(true);

		// Topology run
		log.info("DataFlowTopology start");
		conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("dataflow", conf, builder.createTopology());
		Thread.sleep(1000 * 60 * 5);
		cluster.shutdown();

	}
}